--[[
    原子性有序集合，元素不可重复，每次获取元素时， 该元素将会被删除， 即多次获取元素不会重复。
    集合一共有两个槽位， 一个存储已加入的元素定义为数据槽位、一个存储已获取的元素定义为待确认槽位
    数据槽位: 主要存储通过add方法加入的元素， 槽位数据结构为ZSET, value为元素，score为时间戳, 时间戳主要为get方法进行数据筛选， 因为数据不一定立即获取
    待确认槽位: 主要存储Get方法获取后的元素， 槽位数据结构与数据槽位一样， 时间戳为获取时间。 
    		   该槽位主要用于事务补偿:
    		   1)客户端业务正常处理完成，则调用CONFIRM命令进行删除
               2)处理失败调用REJECT命令将数据回滚到数据槽位
               3)如果过程出现异常，无法与Redis交互，可以调用REJECTBYEXPIRED命令将长时间未处理的数据回滚到数据槽位
]]
--[[在Lua5.2中，unpack被移到table.unpack]]
local unpackMethod = unpack or table.unpack;

local function get()
    --[[获取批次数据，ARGV[1] 批次数据KEY， ARGV[2]: 获取数量， ARGV[3]: 将存在指定毫秒内的数据获取出来， 传0全部获取(ms), ARGV[4]: 当前服务器时间戳(ms) 兼容<3.0版本]]
    local currentTime = tonumber(ARGV[4]);
    local endTime = currentTime - tonumber(ARGV[3]);
    local fetchSize = tonumber(ARGV[2]);
    local confirmKey = ARGV[1] .. '_confirm'
    local result = {};
    local limitForm = 0;
    local limitTo = fetchSize;
    local resultIndex = 1;
    while resultIndex <= fetchSize
    do
        local fetchDatas = redis.call('ZRANGEBYSCORE',ARGV[1], 0, endTime, 'LIMIT', limitForm, limitTo)
        if fetchDatas == nil or #fetchDatas == 0 then
            break
        end
        
        local confirmList = {};
    	local confirmListIndex = 1;
    	
        for i in pairs(fetchDatas) do
            if resultIndex > fetchSize then 
                break
            end
            confirmList[confirmListIndex] = currentTime;
            confirmList[confirmListIndex + 1] = fetchDatas[i];
            result[resultIndex] = fetchDatas[i];
            confirmListIndex = confirmListIndex + 2;
            resultIndex = resultIndex + 1
        end
        redis.call('ZREM',ARGV[1], unpackMethod(fetchDatas));
        redis.call('ZADD',confirmKey, unpackMethod(confirmList));
        
        limitForm = limitTo + 1;
        limitTo = limitForm + fetchSize;
    end

    return result;
end

local function confirm() 
    --[[确认批次数据，ARGV[1] 批次数据KEY， ARGV[2]: 确认的数据]]
    local confirmKey = ARGV[1] .. '_confirm'
    local objectKey = ARGV[1] .. '_object'
    local confirmSize = 0;
    confirmSize = redis.call('ZREM', confirmKey, unpackMethod(ARGV, 2, #ARGV));
    redis.call('HDEL', objectKey, unpackMethod(ARGV, 2, #ARGV));
    return confirmSize;
end

local function reject() 
    --[[驳回批次数据，ARGV[1] 批次数据KEY， ARGV[2]: 当前服务器时间, ARGV[...3]: 确认的数据]]
    local confirmKey = ARGV[1] .. '_confirm'
    local batchKey = ARGV[1];
    local currentTime = tonumber(ARGV[2]);
    local rejectSize = 0;
    for i = 3, #ARGV do
        local delSize = tonumber(redis.call('ZREM', confirmKey, ARGV[i]));
        if delSize == 1 then 
            local addSize = tonumber(redis.call('ZADD', batchKey, currentTime, ARGV[i]));
            if addSize == 1 then 
                rejectSize = rejectSize + 1;
            end
        end
        
    end

    return rejectSize;
end

local function rejectByExpired() 
    --[[驳回过期批次数据，ARGV[1] 批次数据KEY， ARGV[2]: 当前服务器时间, ARGV[3]: 过期时间 (ms)]]
    local confirmKey = ARGV[1] .. '_confirm'
    local batchKey = ARGV[1];
    local currentTime = tonumber(ARGV[2]);
    local expriedTime = tonumber(ARGV[3]);
    local expriedEndTime = currentTime - expriedTime;
    local step = 1000;
    local limitForm = 0;
    local limitTo = step;
    local rejectSize = 0;
    while true do
        local fetchDatas = redis.call('ZRANGEBYSCORE',confirmKey, 0, expriedEndTime, 'LIMIT', limitForm, limitTo)
        if fetchDatas == nil or #fetchDatas == 0 then
            break
        end

        for i in pairs(fetchDatas) do
            local delSize = tonumber(redis.call('ZREM',confirmKey, fetchDatas[i]));
            if delSize == 1 then
                local addSize = tonumber(redis.call('ZADD',batchKey, currentTime, fetchDatas[i]));
                if addSize == 1 then 
                    rejectSize = rejectSize + 1;
                end
            end
           
        end

        if #fetchDatas < step then
            break
        end
        limitForm = limitTo + 1
        limitTo = limitForm + step
    end

    return rejectSize;
end

local function add() 
    local currentTime = tonumber(ARGV[2]);
    local batchKey = ARGV[1];
    local addCount = 0;
    local batchAdd = {};
    local batchAddIndex = 1;
    for i = 3, #ARGV do
    	batchAdd[batchAddIndex] = currentTime;
    	batchAdd[batchAddIndex + 1] = ARGV[i];
    	batchAddIndex = batchAddIndex + 2;
    end
    
    addCount = tonumber(redis.call('ZADD', batchKey, unpackMethod(batchAdd)));
   	return addCount; 
end

local function addObject()
	local batchKey = ARGV[1];
	local currentTime = tonumber(ARGV[2]);
	local data = ARGV[3];
	local dataObject = ARGV[4];
	local objectKey = batchKey .. '_object'
	
	if redis.call('ZADD', batchKey, currentTime, data) == 1 and redis.call('HMSET', objectKey, data, dataObject) then
		return 1;
    end
    
    return 0;
end

local function getObject()
	--[[获取批次数据，ARGV[1] 批次数据KEY， ARGV[2]: 获取数量， ARGV[3]: 将存在指定毫秒内的数据获取出来， 传0全部获取(ms), ARGV[4]: 当前服务器时间戳(ms) 兼容<3.0版本]]
	local datas = get();
	local dataObjects = {};
	local batchKey = ARGV[1];
	local objectKey = batchKey .. '_object';
	for i in pairs(datas) do
		dataObjects[i] = redis.call('HGET', objectKey, datas[i]);
	end
	return dataObjects;
end

local command = string.upper(ARGV[1]);
table.remove(ARGV, 1);
--[[GET,CONFIRM,REJECT,REJECTBYEXPIRED.ADD]]
if command == 'GET' then
    return get();
elseif command == 'CONFIRM' then
    return confirm();
elseif command == 'REJECT' then
    return reject(); 
elseif command == 'REJECTBYEXPIRED' then
    return rejectByExpired(); 
elseif command == 'ADD' then
    return add();
elseif command == 'ADDOBJECT' then
    return addObject();
elseif command == 'GETOBJECT' then
    return getObject();
end

return -1;